package com.yh.csx.etl.scheduler;

import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import javax.annotation.PreDestroy;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.config.CronTask;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.support.CronSequenceGenerator;
import org.springframework.scheduling.support.PeriodicTrigger;
import com.yh.csx.etl.config.EtlDataMappingManager;
import com.yh.csx.etl.worker.EtlWorker;

/***
 * Spring作业动态调度器 使用Spring本身的scheduled来实现任务调度
 * 
 * @author Robin.Wang
 * @date 2020-04-29
 */
public class EtlSpringScheduler implements SchedulingConfigurer {

    private static Logger logger = LoggerFactory.getLogger(EtlSpringScheduler.class);

    @Autowired
    ThreadPoolTaskExecutor taskExecutor;
    private volatile ScheduledTaskRegistrar registrar;
    private final ConcurrentHashMap<String, ScheduledFuture<?>> scheduledFutures = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, CronTask> cronTasks = new ConcurrentHashMap<>();

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        this.registrar = taskRegistrar;
        // 每5s扫描一次任务列表
        this.registrar.addTriggerTask(() -> {
            this.refreshTasks();
        }, triggerContext -> new PeriodicTrigger(5L, TimeUnit.SECONDS).nextExecutionTime(triggerContext));

    }

    /**
     * 刷新任务
     */
    private void refreshTasks() {
        EtlDataMappingManager.loadConfigs("");
        Set<String> taskIds = scheduledFutures.keySet();
        for (String taskId : taskIds) {
            if (EtlDataMappingManager.MAPPINGS.get(taskId) == null) {
                scheduledFutures.get(taskId).cancel(false);
            }
        }
        EtlDataMappingManager.MAPPINGS.forEach((taskId, EtlDataMappingConfig) -> {
            String expression = EtlDataMappingConfig.getCron();
            if (StringUtils.isEmpty(expression) || !CronSequenceGenerator.isValidExpression(expression)) {
                logger.error("定时任务DynamicTask cron表达式不合法: " + expression);
                return;
            }
            // 如果配置一致，则不需要重新创建定时任务
            if (scheduledFutures.containsKey(taskId) && cronTasks.get(taskId).getExpression().equals(expression)) {
                return;
            }
            // 如果策略执行时间发生了变化，则取消当前策略的任务
            if (scheduledFutures.containsKey(taskId)) {
                scheduledFutures.remove(taskId).cancel(false);
                cronTasks.remove(taskId);
            }
            CronTask task = new CronTask(new EtlWorker(EtlDataMappingManager.MAPPINGS.get(taskId)), expression);
            ScheduledFuture<?> future = registrar.getScheduler().schedule(task.getRunnable(), task.getTrigger());
            cronTasks.put(taskId, task);
            scheduledFutures.put(taskId, future);
        });
    }

    @PreDestroy
    public void destroy() {
        this.registrar.destroy();
    }

}
